package com.sh.data.engine.domain.task;

import cn.hutool.core.thread.ThreadUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.common.util.LogArchiverUtil;
import com.sh.data.engine.domain.api.service.DataApiCallRecordService;
import com.sh.data.engine.domain.datadev.offline.service.DatadevJobService;
import com.sh.data.engine.domain.integration.offline.service.OfflineSyncRunningRecordService;
import com.sh.data.engine.domain.integration.online.model.param.StartOnlineKafkaTaskParam;
import com.sh.data.engine.domain.integration.online.service.FlinkDataSyncService;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.infrastructure.config.FlinkConfig;
import com.sh.data.engine.infrastructure.config.LogConfig;
import com.sh.data.engine.job.admin.dao.XxlJobLogDao;
import com.sh.data.engine.job.core.util.CmdUtil;
import com.sh.data.engine.repository.dao.integration.online.entity.FlinkDataSyncTaskEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.net.Socket;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Component
@Slf4j
public class DataEngineAllScheduleTask {

  @Value("${data.engine.logSize:200000}")
  private Integer apiMaxNum;

  @Value("${data.engine.logDays:-7}")
  private Integer apiLogDays;

  @Autowired private DatadevJobService datadevJobService;

  @Autowired private OfflineSyncRunningRecordService offlineSyncRunningRecordService;

  @Autowired private XxlJobLogDao xxlJobLogDao;

  @Autowired private LogConfig logConfig;

  @Autowired private FlinkConfig flinkConfig;

  @Autowired private FlinkDataSyncService flinkDataSyncService;

  @Autowired private DataApiCallRecordService dataApiCallRecordService;

  @Autowired private UserService userService;

  @Scheduled(cron = "0 0 0,12 * * ?")
  public void archiverHufuAllLog() {
    Integer logSize = this.apiMaxNum;
    Integer logDays = this.apiLogDays;
    // 数据服务日志文件
    String apiPath = logConfig.getApiPath();
    LogArchiverUtil.archiveLogsByStrategy(apiPath, logDays, logSize);
    // 离线开发日志文件
    String devJob = logConfig.getOfflineDevJob();
    LogArchiverUtil.archiveLogsByStrategy(devJob, logDays, logSize);
    // 数据集成日志文件
    String offlineSyncJob = logConfig.getOfflineSyncJob();
    LogArchiverUtil.archiveLogsByStrategy(offlineSyncJob, logDays, logSize);

    // 接口采集日志
    String online = logConfig.getOnline();
    LogArchiverUtil.archiveLogsByStrategy(online, logDays, logSize);
    log.info("日志归档已完成");
  }

  /** 虎符所有通过代码定时任务 */
  @Scheduled(cron = "0 0/5 * * * ?")
  public void clearRecord() {
    Calendar cal = Calendar.getInstance();
    cal.add(Calendar.DATE, -1);
    cal.add(Calendar.MONTH, -1);
    // 一个月前的时间
    Date time = cal.getTime();
    log.info("保留一个月的记录");
    try {
      clearXxlLog(time);
    } catch (Exception e) {
      log.error("清除xxl log失败", e.getMessage());
    }

    keepOneMonthRecord(time);

    Calendar cal2 = Calendar.getInstance();

    // 将日期减去15天（半个月）
    cal2.add(Calendar.DATE, -15);
    Date time1 = cal2.getTime();
    keepHalfMonthRecord(time1);
  }

  private void keepHalfMonthRecord(Date time) {
    // 保留半个月的记录
    dataApiCallRecordService.keepHalfMonthRecord(time);
  }

  private static boolean isOlderThanRetentionDays(File file, Integer day) {
    long diff = System.currentTimeMillis() - file.lastModified();
    return TimeUnit.DAYS.convert(diff, TimeUnit.MILLISECONDS) > day;
  }

  /** 主要目的：定时检查flink是否启动，挂了重新拉起 */
  @Scheduled(cron = "0 0/5 * * * ?")
  public void checkFlinkStatus() {

    String flinkHome = flinkConfig.getHome();
    String port = flinkConfig.getPort();
    String cmd = flinkHome + "start-cluster.sh";
    try {
      // FIXME 只检测JobManager服务器节点
      Socket localhost = new Socket("localhost", Integer.parseInt(port));
    } catch (IOException e) {
      try {
        // runCmd("root",args.toArray(new String[args.size()]));
        CmdUtil.runCmd(cmd);
      } catch (Exception exception) {
        log.info("启动flink服务失败");
      }
    }

    String[] result = new String[40];
    String listCmd = flinkHome + "flink list";
    try {
      result = CmdUtil.runCmd(listCmd);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }

    log.info("checkFlinkStatus result" + result.toString());
    List<String> list1 = new ArrayList<>();
    if (result != null && result.length > 0) {
      String pattern = "\\b[0-9a-f]{32}\\b";
      Pattern r = Pattern.compile(pattern);
      for (String line : result) {
        Matcher m = r.matcher(line);
        while (m.find()) {
          list1.add(m.group().trim());
        }
      }
    }
    log.info("checkFlinkStatus" + list1.toString());
    List<FlinkDataSyncTaskEntity> list =
        flinkDataSyncService.list(
            new LambdaQueryWrapper<>(FlinkDataSyncTaskEntity.class)
                .eq(FlinkDataSyncTaskEntity::getTaskStatus, 2)
                .isNotNull(FlinkDataSyncTaskEntity::getFlinkJobId));
    for (FlinkDataSyncTaskEntity flinkDataSyncTaskEntity : list) {
      String flinkJobId = flinkDataSyncTaskEntity.getFlinkJobId();
      if (list1.contains(flinkJobId.trim())) {
        flinkDataSyncTaskEntity.setTaskRunStatus(OfflineSyncTaskStatus.running.getTaskStatus());
      } else {
        flinkDataSyncTaskEntity.setTaskRunStatus(OfflineSyncTaskStatus.failure.getTaskStatus());
      }
    }
    flinkDataSyncService.updateBatchById(list);

    // 启动flink 任务
    flinkJobMonitor();
  }

  void clearXxlLog(Date time) {
    xxlJobLogDao.cleanRecord(time);
  }

  private void keepOneMonthRecord(Date time) {

    // clear job history
    datadevJobService.cleanMonthAgoRecord(time);

    // clear offline sync
    offlineSyncRunningRecordService.clearAMonthAgoRecord(time);
  }

  public void flinkJobMonitor() {
    List<FlinkDataSyncTaskEntity> tasks = flinkDataSyncService.getAllPublishTask();

    Map<String, FlinkDataSyncTaskEntity> relaseJobMap =
        tasks.stream()
            .filter(task -> StringUtils.isNotBlank(task.getFlinkJobId()))
            //            .map(FlinkDataSyncTaskEntity::getFlinkJobId)
            .collect(Collectors.toMap(FlinkDataSyncTaskEntity::getFlinkJobId, Function.identity()));
    Set<String> relaseJobIds = relaseJobMap.keySet();
    // ArrayList<String> args = Lists.newArrayList();
    String flinkHome = flinkConfig.getHome();
    //    args.add(flinkHome+"flink");
    //    args.add("list");
    String cmd = flinkHome + "flink list";
    Integer startIndex = 0;
    Integer endIndex = 0;
    List<String> flink = Lists.newArrayList();
    List<String> runningStr = Lists.newArrayList();
    List<String> runningJobIds = Lists.newArrayList();
    try {
      // flink = runCmd("root", args.toArray(new String[args.size()]));
      String[] result = CmdUtil.runCmd(cmd);
      flink = Arrays.stream(result).collect(Collectors.toList());
      for (int i = 0; i < flink.size(); i++) {
        String line = flink.get(i);
        int i1 = line.indexOf("------------------ Running/Restarting Jobs -------------------");
        int i2 = line.indexOf("--------------------------------------------------------------");
        if (i1 != -1) {
          startIndex = i;
        }
        if (i2 != -1) {
          endIndex = i;
        }
      }
    } catch (Exception e) {

    }
    if (startIndex > 0 && endIndex > startIndex) {
      runningStr = flink.subList(startIndex + 1, endIndex);
    }
    runningStr.forEach(
        i -> {
          String[] split = i.split(":");
          runningJobIds.add(split[3].trim());
        });

    relaseJobIds.removeAll(runningJobIds);
    if (CollectionUtils.isNotEmpty(relaseJobIds)) {
      relaseJobIds.forEach(
          jobId -> {
            FlinkDataSyncTaskEntity flinkDataSyncTaskEntity = relaseJobMap.get(jobId);
            StartOnlineKafkaTaskParam param = new StartOnlineKafkaTaskParam();
            param.setTaskId(flinkDataSyncTaskEntity.getId());
            flinkDataSyncService.publish(param);
            ThreadUtil.sleep(2000);
          });
    }
  }

  public static void main(String[] args) {

    Calendar cal = Calendar.getInstance();

    // 将日期减去15天（半个月）
    cal.add(Calendar.DATE, -15);

    // 打印半个月前的日期
    System.out.println("半个月前的日期: " + cal.getTime());
    String[] lines = {
      "Waiting for response...",
      "------------------ Running/Restarting Jobs -------------------",
      "05.06.2024 16:12:31 : e3a1e648c390d4465e99611b0e0d9da6 : insert-into_default_catalog.default_database.ods_100w_target (RUNNING)",
      "05.06.2024 16:12:37 : e617f61c5d31c8833a9a171b3896da12 : insert-into_default_catalog.default_database.500w_target (RUNNING)",
      "05.06.2024 16:12:43 : e85197e3c706c291edd22561f9792b5b : insert-into_default_catalog.default_database.newtable_target (RUNNING)",
      "05.06.2024 17:41:33 : 6ab357e81500b2e5acf615b90adc0229 : insert-into_default_catalog.default_database.ods_USERS_target (RUNNING)",
      "--------------------------------------------------------------",
      "No scheduled jobs."
    };

    // 正则表达式模式
    String pattern = "\\b[0-9a-f]{32}\\b";
    Pattern r = Pattern.compile(pattern);

    List<String> result = new ArrayList<>();

    for (String line : lines) {
      Matcher m = r.matcher(line);
      while (m.find()) {
        result.add(m.group().trim());
      }
    }

    // 打印结果
    for (String str : result) {
      System.out.println(str);
    }
  }
}
